#include "redismanager.h"
#include <unistd.h>
#include <sys/time.h>
#include "redis++.h"

using namespace sw;
using namespace sw::redis;

RedisManager::RedisManager()
{

}

RedisManager::~RedisManager()
{
    stopTask();
}

RedisManager *RedisManager::getInstance()
{
    static RedisManager manager;
    return &manager;
}

void RedisManager::initRedis(const STU_REDIS_CONFIG &config, const STU_REDIS_CHANNEL &channel)
{
    m_config = config;
    m_channel = channel;
}

void RedisManager::startTask()
{
    if (nullptr == m_redis)
    {
        ConnectionOptions opts;
        opts.host = m_config.redis_ip;
        opts.port = m_config.redis_port;
        opts.db = m_config.default_db;
        opts.connect_timeout = std::chrono::milliseconds(3000);
        opts.socket_timeout = std::chrono::milliseconds(3000);

        ConnectionPoolOptions pool_options;
        pool_options.size = 10;

        try
        {
            m_connected = false;

            log_printf(LOG_LEV, "RedisManager ip %s %d", m_config.redis_ip.c_str(), m_config.redis_port);
            m_redis = new redis::Redis(opts, pool_options);
            if(m_redis != nullptr)
            {
                //ping可能造成长时间阻塞 设置上方超时
                string info = m_redis->ping();
                if(string::npos != info.find("PONG"))
                {
                    m_connected = true;
                    log_printf(LOG_LEV, "Redis Connect Sucess! ping %s", info.c_str());
                }
            }
        }
        catch (const std::exception& e)
        {
            m_connected = false;
            log_printf(LOG_LEV, "RedisManager exception %s", e.what());

            if(m_redis != nullptr)
            {
                delete m_redis;
                m_redis = nullptr;
            }
        }
    }

    if(!m_connected)
    {
        return;
    }

    m_bRun = true;

    if (m_thread == nullptr)
    {
        m_thread = new std::thread(&RedisManager::loopRun, this);
    }

    if (m_subThread == nullptr)
    {
        m_subThread = new std::thread(&RedisManager::subLoopRun, this);
    }
}

void RedisManager::stopTask()
{
    m_bRun = false;
    m_cv.notify_all();

    if(m_thread != nullptr )
    {
        if(m_thread->joinable())
        {
            m_thread->join();
        }

        delete m_thread;
        m_thread = nullptr;
    }

    if(m_subThread != nullptr )
    {
        if(m_subThread->joinable())
        {
            m_subThread->join();
        }

        delete m_subThread;
        m_subThread = nullptr;
    }

    if (nullptr != m_redis)
    {
        try
        {
            m_redis->command("quit");
            delete m_redis;
            m_redis = nullptr;
            log_printf(LOG_LEV, "RedisManager quit!");
        }
        catch (const Error &err)
        {
//            log_printf(LOG_LEV, "RedisManager quit %s", err.what());
        }
    }
}

void RedisManager::autoReconnect()
{
    if(!m_connected)
    {
        log_printf(LOG_LEV, "autoReconnect %s:%d", m_config.redis_ip.c_str(), m_config.redis_port);
        stopTask();
        startTask();
    }
}

void RedisManager::loopRun()
{
    log_printf(LOG_LEV, "RedisManager loopRun!");

    std::unique_lock<std::mutex> lck(m_mutex);
    while (m_bRun)
    {
        m_cv.wait(lck);

        //防止redis连接失败 数据交互崩溃问题
        if(!m_connected)
        {
            log_printf(LOG_LEV, "RedisManager redis not connected!");
            continue;
        }

        if (!m_vecAddFiles.empty())
        {
            for (auto info : m_vecAddFiles)
            {
                int len = m_redis->lpush(m_channel.keyAddFile.c_str(), info.c_str());
                log_printf(LOG_LEV, "redis addFile! %s %d", info.c_str(), len);
            }
            m_vecAddFiles.clear();

            m_redis->publish(m_channel.mediaServiceCmdChannel.c_str(), m_channel.cmdAddFile);
        }
        if (!m_vecDeleteFiles.empty())
        {
            for (auto info : m_vecDeleteFiles)
            {
                int len = m_redis->lpush(m_channel.keyDelFile.c_str(), info.c_str());
                log_printf(LOG_LEV, "redis delFile! %s %d", info.c_str(), len);
            }
            m_vecDeleteFiles.clear();

            m_redis->publish(m_channel.mediaServiceCmdChannel.c_str(), m_channel.cmdDelFile);
        }
    }
}

void RedisManager::subLoopRun()
{
    log_printf(LOG_LEV, "RedisManager subLoopRun!");

    try
    {
        //Create a Subscriber.
        Subscriber sub = m_redis->subscriber();
        //Subscribe to channels and patterns.
        sub.subscribe(m_channel.controllerCmdChannel);
        log_printf(LOG_LEV, "Subscriber channel %s", m_channel.controllerCmdChannel.c_str());

        //消息回调
        sub.on_message([=](std::string channel, std::string msg)
        {
            if (channel == m_channel.controllerCmdChannel)
            {
                log_printf(LOG_LEV, "Subscriber message %s %s", channel.c_str(), msg.c_str());

                if(msg == m_channel.cmdStorage)
                {
                    //nothing
                    log_printf(LOG_LEV, "nothing");
                }
                else if(msg == m_channel.cmdDatetimeUpdate)
                {
                    //更新系统时间
                    std::string dateTime = m_redis->get(m_channel.keyDatetimeUpdate).value();
                    updateSystemDatetime(dateTime);
                }
            }
        });

        while (m_bRun)
        {
            try
            {
                //阻塞等待 有消息则回调
                sub.consume();
            }
            catch (const sw::redis::TimeoutError)
            {
                if (!m_bRun)
                {
                    sub.unsubscribe(m_channel.controllerCmdChannel);
                    return;
                }
            }
            catch (const Error &err)
            {
                log_printf(LOG_LEV, "sub.consume() %s", err.what());
            }
        }
    }
    catch (const std::exception& e)
    {
        log_printf(LOG_LEV, "Create a Subscriber %s", e.what());
    }
}

void RedisManager::addFile(const string &jsonInfo)
{
    std::unique_lock<std::mutex> lck(m_mutex);
    m_vecAddFiles.push_back(jsonInfo);
    m_cv.notify_all();
}

void RedisManager::delFile(const string &filePath)
{
    std::unique_lock<std::mutex> lck(m_mutex);
    m_vecDeleteFiles.push_back(filePath);
    m_cv.notify_all();
}

void RedisManager::updateSystemDatetime(const string &dateTime)
{
    log_printf(LOG_LEV, "UpdateSystemDatetime %s", dateTime.c_str());
    if(dateTime.empty())
    {
        return;
    }

    std::string command = "sudo date -s \"" + dateTime + "\"";
    int ret = system(command.c_str());
    if(ret != 0)
    {
        log_printf(LOG_LEV, "UpdateSystemDatetime error %s", command.c_str());
    }
}

